package com.atguigu.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventLog}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.util.control.Breaks._
object AlertApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AlertApp")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf,Seconds(5))

    //3.消费kafka数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT,ssc)

    //4.将读取到的json字符串转为样例类,并转为K，V
    val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
    val midToEventLogDStream = kafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])

        val times: String = sdf.format(new Date(eventLog.ts))

        eventLog.logDate = times.split(" ")(0)
        eventLog.logHour = times.split(" ")(1)

        (eventLog.mid,eventLog)
      })
    })

    //5.开启一个5min窗口
    val windowMidToLogDStream: DStream[(String, EventLog)] = midToEventLogDStream.window(Minutes(5))

    //6.对相同mid的数据做聚合
    val midToIterLogDStream: DStream[(String, Iterable[EventLog])] = windowMidToLogDStream.groupByKey()

    //7.根据条件筛选数据，生成疑似预警日志
    /**
      * 同一设备，（再做，groupBykey）
      * 5分钟内 （先，开启5min的窗口）
      * 三次及以上用不同账号登录并领取优惠劵，（先判断用户是否有领优惠券，然后对领优惠券的去重（set集合去重），去重完毕后不同账号的个数大于等于三（计算       Set集合大小是否大于等于三））
      * 并且过程中没有浏览商品。（判断用户是否浏览商品，如果浏览商品，则不符合以上要求，可以直接过滤）
      */
    val boolToCouponAlertInfoDStream: DStream[(Boolean, CouponAlertInfo)] = midToIterLogDStream.mapPartitions(partition => {
      partition.map { case (mid, iter) =>
        //创建用来存放领优惠券的用户id的set集合
        val uids: util.HashSet[String] = new util.HashSet[String]()
        //创建用来存放领优惠券所涉及的商品id的Set集合
        val itemIds: util.HashSet[String] = new util.HashSet[String]()
        //创建用来存放用户涉及行为的list集合
        val events: util.ArrayList[String] = new util.ArrayList[String]()

        //创建一个标志位，用来判断5分钟内用户是否有浏览商品行为
        var bool: Boolean = true
        breakable {
          iter.foreach(log => {
            events.add(log.evid)
            //判断5min钟内用户是否有浏览商品行为，有浏览商品行为则跳出循环
            if ("clickItem".equals(log.evid)) {
              //有浏览商品行为，则不符合预警要求，跳出循环
              bool = false
              break()
            } else if ("coupon".equals(log.evid)) {
              //没有浏览商品但是领优惠券
              uids.add(log.uid)
              itemIds.add(log.itemid)
            }
          })
        }
        //生成疑似预警日志
        (uids.size() >= 3 && bool, CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis()))
      }
    })
    
    //8.生成预警日志
    val couponAlertInfo: DStream[CouponAlertInfo] = boolToCouponAlertInfoDStream.filter(_._1).map(_._2)
    couponAlertInfo.print()

    //9.将预警日志写入ES
    /**
      * 达到以上要求则产生一条预警日志。
      * 并且同一设备，每分钟只记录一次预警（对预警信息去重，ES幂等性去重->docId怎么设置？mid+精确到分钟的时间）
      */
    couponAlertInfo.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, CouponAlertInfo)] = partition.toList.map(log => {
          (log.mid + log.ts / 1000 / 60, log)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_ALERT_INDEX+"211227",list)
      })
    })

    //10.启动任务并阻塞
    ssc.start()
    ssc.awaitTermination()
  }

}
